package kafka

import (
	"Gin-Gorm-boke/boke_common/models"
	"context"
	"github.com/segmentio/kafka-go"
	"gorm.io/gorm"

	"log"
)

const (
	consumerKafkaBroker = "localhost:9092"    // 替换为你的 Kafka 地址
	consumerKafkaTopic  = "article_viewCount" // 替换为你的 Kafka topic
	consumerGroupID     = "article"           // 替换为你的消费者组 ID
)

func StartConsumer() error {
	reader := kafka.NewReader(kafka.ReaderConfig{
		Brokers:     []string{consumerKafkaBroker},
		Topic:       consumerKafkaTopic,
		GroupID:     consumerGroupID,
		StartOffset: kafka.FirstOffset,
	})

	defer reader.Close()

	log.Printf("Kafka consumer started, listening on topic %s\n", consumerKafkaTopic)

	for {
		msg, err := reader.ReadMessage(context.Background())
		if err != nil {
			return err
		}

		log.Printf("Received message: key=%s value=%s", string(msg.Key), string(msg.Value))
		id := string(msg.Value)
		models.Db.Model(new(models.Article)).Where("id = ?", id).Update("view_count", gorm.Expr("view_count+1"))
	}
}
